package org.rency.crawler.integration.queue;

import javax.annotation.Resource;
import org.rency.common.utils.converts.JsonConvert;
import org.rency.common.utils.domain.BaseResult;
import org.rency.crawler.common.enums.QueueKind;
import org.rency.crawler.domain.queue.FetchQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

/**
 * 描述: 待抓取队列<br>
 * 最近修改时间:2017/11/29 15:15<br>
 *
 * @author CaiyuRen
 * @version V1.0
 * @since 2017/11/29 15:15
 */
@Component("fetchQueueClient")
public class FetchQueueClient implements QueueClient<FetchQueue>{

  private static final Logger logger = LoggerFactory.getLogger(FetchQueueClient.class);

  @Resource
  private KafkaTemplate<String, String> kafkaTemplate;

  @Override
  public BaseResult publish(final FetchQueue queue) {
    logger.debug("发布待抓取队列:{}",queue);
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(
        QueueKind.FETCH_QUEUE.getQueue(),JsonConvert.toJson(queue));
    future.addCallback(new SuccessCallback<SendResult<String, String>>() {
      @Override
      public void onSuccess(SendResult<String, String> stringFetchQueueSendResult) {
        logger.debug("待抓取队列发布成功:{}",stringFetchQueueSendResult.getProducerRecord());
      }
    }, new FailureCallback() {
      @Override
      public void onFailure(Throwable throwable) {
        logger.warn("待抓取队列发布失败:{}",queue,throwable);
      }
    });
    return BaseResult.success("发布成功");
  }
}
